package com.zhisheng.sql.blink.stream.example;

import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.HashMap;

/**
 * CasinoWinJob
 *
 * @author yongxiang
 * @date 2025/2/28 14:59
 * @desc
 */
public class TestJob {
    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
        TableConfig config = tableEnv.getConfig();
        config.getConfiguration().setBoolean("table.exec.emit.early-fire.enabled",true);
        config.getConfiguration().setString("table.exec.emit.early-fire.delay","5s");
        // 定义 Debezium Kafka 源表
        tableEnv.executeSql(
                "CREATE TABLE person_click_records ("
                        + "  id BIGINT,"
                        + "  person_id INT,"
                        + "  click_time BIGINT, "  // 原始字段为 BIGINT（Unix 时间戳，毫秒）
                        + "  ts AS TO_TIMESTAMP_LTZ(click_time - 8 * 3600 * 1000, 3), "  // 转换为 TIMESTAMP
                        + "  click_num INT,"
                        + "  WATERMARK FOR ts AS ts - INTERVAL '5' SECOND"
                        + ") WITH ("
                        + "  'connector' = 'kafka',"
                        + "  'topic' = 'mysql-server.test_sm.person_click_records',"
                        + "  'properties.bootstrap.servers' = '192.168.8.61:9092,192.168.8.61:9093,192.168.8.61:9094',"
                        + "  'properties.group.id' = 'click_consumer',"
                        + "  'scan.startup.mode' = 'latest-offset',"
                        + "  'format' = 'debezium-json',"
                        + "  'debezium-json.schema-include' = 'true'"
                        + ")");

        Table result = tableEnv.sqlQuery(
                "SELECT "
                        + "  person_id, "
                        + "  TUMBLE_START(ts, INTERVAL '1' MINUTE) AS window_start, "
                        + "  TUMBLE_END(ts, INTERVAL '1' MINUTE) AS window_end, "
                        + "  SUM(click_num) AS total_clicks "
                        + "FROM person_click_records "
                        + "GROUP BY "
                        + "  person_id, "
                        + "  TUMBLE(ts, INTERVAL '1' MINUTE)"
        );
        // 4. 输出中间结果（调试用）
        Table personAppleTable = tableEnv.from("person_click_records");
        tableEnv.toChangelogStream(personAppleTable).print("person_click_records");
        tableEnv.toChangelogStream(result).print("hello");
        env.execute("Real-time Click Analytics");


    }


}
